package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"go-canal/utils/kafka"
	"time"
)

func customer(consumerMsg *sarama.ConsumerMessage) error {
	fmt.Printf("topic=%+v,position=%+v,offset=%+v, msg=%+v\n", consumerMsg.Topic, consumerMsg.Partition, consumerMsg.Offset, string(consumerMsg.Value))
	time.Sleep(time.Second)
	fmt.Printf("提交offset\n")
	return nil
}

func main() {
	//group := "dynamics_analysis"
	//topics := []string{"account", "broker_config", "order_info", "order_swap", "contract_info", "business_transaction",
	//	"currency_info", "currency_config", "middle_usercollect"}
	group := "analysis"
	topics := []string{"analysis"}
	consumer := kafka.NewDefaultConsumer(group, []string{"192.168.200.142:9092"}, topics, customer)

	if err := consumer.Start(); err != nil {
		panic(err)
	}
	for {

	}
}
